package org.example.queueadmin.task.runner;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.example.queueadmin.dlq.DeadLetterService;
import org.example.queueadmin.log.MessageLogService;
import org.example.queueadmin.task.TaskRunner;
import org.example.queueadmin.task.TransferTask;
import org.example.queueadmin.task.TransferTaskRepository;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@RequiredArgsConstructor
public class KafkaToRabbitRunner implements TaskRunner {
  private final TransferTask task;
  private final DeadLetterService dlq;
  private final TransferTaskRepository repository;
  private final MessageLogService messageLogService;
  private KafkaMessageListenerContainer<byte[], byte[]> container;
  private RabbitTemplate rabbitTemplate;
  private final AtomicBoolean running = new AtomicBoolean(false);

  @Override
  public synchronized void start() {
    if (running.get()) return;
    running.set(true);

    Map<String, Object> props = new HashMap<>();
    props.put("bootstrap.servers", task.getKafkaSourceBootstrap());
    props.put("group.id", task.getKafkaSourceGroup());
    props.put("enable.auto.commit", false);
    props.put("auto.offset.reset", task.getKafkaSourceOffsetReset() == null ? "latest" : task.getKafkaSourceOffsetReset());
    props.put("key.deserializer", ByteArrayDeserializer.class);
    props.put("value.deserializer", ByteArrayDeserializer.class);

    // 应用重试配置
    props.put("retries", task.getMaxRetries() != null ? task.getMaxRetries() : 3);
    props.put("retry.backoff.ms", task.getRetryBackoffMs() != null ? task.getRetryBackoffMs() : 1000);

    DefaultKafkaConsumerFactory<byte[], byte[]> cf = new DefaultKafkaConsumerFactory<>(props);
    ContainerProperties containerProps = new ContainerProperties(task.getKafkaSourceTopic());
    containerProps.setAckMode(ContainerProperties.AckMode.MANUAL);
    containerProps.setMessageListener((org.springframework.kafka.listener.AcknowledgingMessageListener<byte[], byte[]>) (record, ack) -> {
      String source = task.getKafkaSourceTopic() + ":" + record.partition() + ":" + record.offset();
      String target = task.getRabbitTargetExchange() + "/" + task.getRabbitTargetRoutingKey();

      try {
        ensureRabbitTemplate();
        rabbitTemplate.convertAndSend(task.getRabbitTargetExchange(), task.getRabbitTargetRoutingKey(), record.value());
        ack.acknowledge();

        // 记录成功日志
        messageLogService.logMessage(task.getId(), "Kafka→Rabbit", source, target, record.value(), true, null);

        // 更新成功处理计数
        updateProcessedCount();
      } catch (Exception e) {
        log.error("Kafka->Rabbit 转发失败, task={}, error={}", task.getName(), e.getMessage(), e);
        dlq.push(task.getId(), record.value(), e.getMessage());

        // 记录失败日志
        messageLogService.logMessage(task.getId(), "Kafka→Rabbit", source, target, record.value(), false, e.getMessage());

        // 更新失败计数和错误信息
        updateFailedCount(e.getMessage());
      }
    });

    container = new KafkaMessageListenerContainer<>(cf, containerProps);
    container.start();
  }

  private void ensureRabbitTemplate() {
    if (rabbitTemplate != null) return;
    CachingConnectionFactory cf = new CachingConnectionFactory(task.getRabbitTargetHost(), task.getRabbitTargetPort());
    cf.setUsername(task.getRabbitTargetUsername());
    cf.setPassword(task.getRabbitTargetPassword());
    cf.setVirtualHost(task.getRabbitTargetVhost());
    rabbitTemplate = new RabbitTemplate(cf);
    rabbitTemplate.setChannelTransacted(false);

    // 应用重试配置
    if (task.getMaxRetries() != null && task.getMaxRetries() > 0) {
      rabbitTemplate.setRetryTemplate(createRetryTemplate());
    }
    rabbitTemplate.setMandatory(true);
  }

  @Override
  public synchronized void stop() {
    running.set(false);
    if (container != null) {
      try {
        log.info("Stopping Kafka container for task: {}", task.getName());
        container.stop();
        container = null;
        log.info("Kafka container stopped for task: {}", task.getName());
      } catch (Exception e) {
        log.warn("Error stopping Kafka container for task {}: {}", task.getName(), e.getMessage());
        container = null;
      }
    }

    // 清理RabbitMQ连接
    if (rabbitTemplate != null && rabbitTemplate.getConnectionFactory() instanceof CachingConnectionFactory cf) {
      try {
        cf.destroy();
        rabbitTemplate = null;
      } catch (Exception e) {
        log.warn("Error closing RabbitMQ connection for task {}: {}", task.getName(), e.getMessage());
      }
    }
  }

  @Override
  public synchronized void restart() { stop(); start(); }

  @Override
  public String status() {
    return container != null && running.get() ? "RUNNING" : "STOPPED";
  }

  private void updateProcessedCount() {
    try {
      TransferTask t = repository.findById(task.getId()).orElse(null);
      if (t != null) {
        t.setProcessed((t.getProcessed() == null ? 0 : t.getProcessed()) + 1);
        repository.save(t);
      }
    } catch (Exception e) {
      log.warn("Failed to update processed count: {}", e.getMessage());
    }
  }

  private void updateFailedCount(String errorMsg) {
    try {
      TransferTask t = repository.findById(task.getId()).orElse(null);
      if (t != null) {
        t.setFailed((t.getFailed() == null ? 0 : t.getFailed()) + 1);
        t.setLastErrorAt(Instant.now());
        t.setLastErrorMsg(errorMsg);
        repository.save(t);
      }
    } catch (Exception e) {
      log.warn("Failed to update failed count: {}", e.getMessage());
    }
  }

  private RetryTemplate createRetryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();

    // 重试策略
    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(task.getMaxRetries() != null ? task.getMaxRetries() : 3);
    retryTemplate.setRetryPolicy(retryPolicy);

    // 退避策略
    FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
    backOffPolicy.setBackOffPeriod(task.getRetryBackoffMs() != null ? task.getRetryBackoffMs() : 1000);
    retryTemplate.setBackOffPolicy(backOffPolicy);

    return retryTemplate;
  }
}

